package concurrent.pool;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author Jay
 */
public class BlockingQueue<T> extends ReentrantLock {

    private final Deque<T> queue;
    private final int capacity;

    private final Condition fullWaitSet = newCondition();
    private final Condition emptyWaitSet = newCondition();

    public BlockingQueue(int capacity) {
        // 采用非公平锁
        super(false);
        this.capacity = capacity;
        this.queue = new ArrayDeque<>(capacity);
    }

    /**
     * 无超时等待获取，不建议使用
     * @return 获取结果
     */
    public T take(){
        lock();
        try{
            // 队列为空时，再空等待集合无限时等待
            while(queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }
        finally {
            unlock();
        }
    }

    public void put(T task){
        lock();
        try{
            while(queue.size() == capacity){
                try{
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(task);
            emptyWaitSet.signal();
        }
        finally {
            unlock();
        }
    }

    /**
     * 超时获取，等待时间超时返回null
     * @param timeout 超时时间
     * @param timeUnit 时间单位
     * @return 获取结果
     */
    public T poll(long timeout, TimeUnit timeUnit){
        lock();
        // 等待时间转换成nanos
        long nanos = timeUnit.toNanos(timeout);
        try{
            while(queue.isEmpty()){
                try {
                    // 等待超时，获取失败，返回null
                    if(nanos <= 0) {
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }
        finally {
            unlock();
        }
    }

    /**
     * 超时添加，如果等待时间超时队列仍无空位，添加失败返回false
     * @param task 添加的任务
     * @param timeout 超时时间
     * @param timeUnit 时间单位
     * @return 是否添加成功
     */
    public boolean offer(T task, long timeout, TimeUnit timeUnit){
        lock();
        long nanos = timeUnit.toNanos(timeout);
        try{
            while(queue.size() == capacity){
                try{
                    // 如果等待时间超时，返回false
                    if(nanos <= 0) {
                        return false;
                    }
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        }
        finally {
            unlock();
        }
    }

    /**
     * 根据拒绝策略尝试添加
     * @param task 任务
     * @param rejectPolicy 拒绝策略
     */
    public void tryPut(T task, RejectPolicy<T> rejectPolicy){
        lock();
        try{
            if(queue.size() == capacity){
                rejectPolicy.reject(this, task);
            }
            else{
                put(task);
            }
        }
        finally {
            unlock();
        }
    }

    /**
     * 删除队列第一个任务,unsafe
     */
    public void removeFirst(){
        queue.removeFirst();
    }

    /**
     * 添加到末尾，unsafe
     * @param task 添加到末尾的任务
     */
    public void addLast(T task){
        queue.addLast(task);
    }
}
